/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.processing.split.reducers;

import aiaudio.Application;
import aiaudio.database.tables.ArtistsListTable;
import aiaudio.database.tables.RatingsListTable;
import aiaudio.database.tables.TemporaryTable;
import java.io.IOException;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableReducer;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.BooleanWritable;

/**
 *
 * @author nastya
 */
public class ArtistCountCheckerReducer extends TableReducer<ImmutableBytesWritable, BooleanWritable, ImmutableBytesWritable> {

    public static final String INITIAL_RATING_TABLE_NAME_PARAM = "tableRatings";
    HTable initialRatingMatrix;
    public static final String SPLITTING_COEFFICIENT_PARAM = "splittingCoefficient";
    private double splittingCoefficient;
    public static final String USERS_TABLE_NAME_PARAMETER = "artistsTableName";
    HTable usersTable;

    protected long calculateUsersToCopy(String userName) throws IOException {
        long cnt = getVotingCountByUser(Bytes.toBytes(userName));
        long userToWrite = (long) (cnt * splittingCoefficient);
        if (userToWrite < 0) {
            userToWrite = 1;
        }
        return userToWrite;
    }

    protected void copyToResultTable(Result result, Context context) throws InterruptedException, IOException {
        Put put = new Put(result.getRow());

        String userB = RatingsListTable.getUser(result);
        String artistB = RatingsListTable.getArtist(result);
        int valueB = RatingsListTable.getListenCount(result);

        RatingsListTable.addArtist(put, artistB);
        RatingsListTable.addUser(put, userB);
        RatingsListTable.addListenCount(put, valueB);

        context.write(new ImmutableBytesWritable(result.getRow()), put);
    }

    @Override
    protected void reduce(ImmutableBytesWritable key, Iterable<BooleanWritable> values, Context context) throws IOException, InterruptedException {

        boolean ok = isNotNeedToFix(values);
        if (ok) {
            return;
        }
        long userToWrite = calculateUsersToCopy(Bytes.toString(key.get()));
        Put put = new Put(key.get());
        TemporaryTable.addCount(put, userToWrite);
    }

    protected boolean isNotNeedToFix(Iterable<BooleanWritable> values) {
        for (BooleanWritable booleanWritable : values) {
            if (booleanWritable.get()) {
                return true;
            }
        }
        return false;
    }

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        super.setup(context);
        String tableToCheckName = context.getConfiguration().get(INITIAL_RATING_TABLE_NAME_PARAM);
        initialRatingMatrix = Application.database().getByName(tableToCheckName).getTable();
        String usersTableToCheckName = context.getConfiguration().get(USERS_TABLE_NAME_PARAMETER);
        usersTable = Application.database().getByName(usersTableToCheckName).getTable();
        splittingCoefficient = context.getConfiguration().getFloat(SPLITTING_COEFFICIENT_PARAM, 0.5f);
    }

    protected long getVotingCountByUser(byte[] key) throws IOException {
        return ArtistsListTable.getVotingCount(usersTable, key);
    }

    private void addNonExistsing(String userName, Context context) throws IOException, InterruptedException {
        Scan scan = new Scan();
        scan.setFilter(RatingsListTable.createByUserFilter(userName));
        ResultScanner scanner = initialRatingMatrix.getScanner(scan);
        long userToWrite = calculateUsersToCopy(userName);


        Result result = null;
        long i = 0;
        while ((result = scanner.next()) != null) {
            if (i >= userToWrite) {
                break;
            }
            System.out.println("i = " + i);
            copyToResultTable(result, context);

            i++;
        }
        scanner.close();
    }
}
